package com.example.cdc.factory;

import com.example.cdc.annotation.CdcListener;
import com.example.cdc.constant.CdcConstant;
import com.example.cdc.listener.CdcMessageListener;
import com.example.server.service.SysServerService;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.ApplicationContext;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author chentao
 * @version 1.0
 * @description: TODO
 * @date 2025/3/27 下午8:24
 */
@Service
public class SysCdcFactory implements BeanPostProcessor, InitializingBean, DisposableBean {

    Logger logger = LoggerFactory.getLogger(getClass().getName());

    private Map<String, String> listenerMap = new ConcurrentHashMap<>();

    private Map<Long, String> tableMap = new ConcurrentHashMap<>();

    @Value("${spring.datasource.host}")
    private String host;

    @Value("${spring.datasource.port}")
    private int port;

    @Value("${spring.datasource.db}")
    private String db;

    @Value("${spring.datasource.username}")
    private String username;

    @Value("${spring.datasource.password}")
    private String password;

    private long serverId;

    private BinaryLogClient binaryLogClient;

    private ExecutorService executorService;

    @Resource
    private SysServerService sysServerService;

    @Resource
    private ApplicationContext applicationContext;

    @Override
    public void destroy() throws Exception {
        this.binaryLogClient.disconnect();
        this.executorService.shutdownNow();
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        this.executorService = Executors.newCachedThreadPool();
        this.serverId = sysServerService.getCurrentServerId();
        this.binaryLogClient = new BinaryLogClient(this.host, this.port, this.db, this.username, this.password);
        this.binaryLogClient.setServerId(serverId);
        this.binaryLogClient.registerEventListener(new BinaryLogClient.EventListener() {
            @Override
            public void onEvent(Event event) {
                if (event.getData() instanceof TableMapEventData) {
                    TableMapEventData eventData = event.getData();
                    long tableId = eventData.getTableId();
                    String tableName = eventData.getTable();
                    tableMap.putIfAbsent(tableId, tableName);
                } else {
                    if (event.getData() instanceof UpdateRowsEventData) {
                        //更新
                        UpdateRowsEventData updateRowsEventData = event.getData();
                        String tableName = tableMap.get(updateRowsEventData.getTableId());
                        executorService.execute(new CdcChangeThread(tableName, CdcConstant.UPDATE));
                    }

                    if (event.getData() instanceof WriteRowsEventData) {
                        //新增
                        WriteRowsEventData writeRowsEventData = event.getData();
                        String tableName = tableMap.get(writeRowsEventData.getTableId());
                        executorService.execute(new CdcChangeThread(tableName, CdcConstant.ADD));
                    }

                    if (event.getData() instanceof DeleteRowsEventData) {
                        //删除
                        DeleteRowsEventData deleteRowsEventData = event.getData();
                        String tableName = tableMap.get(deleteRowsEventData.getTableId());
                        executorService.execute(new CdcChangeThread(tableName, CdcConstant.DELETE));
                    }
                }
            }
        });
        this.binaryLogClient.registerLifecycleListener(new BinaryLogClient.AbstractLifecycleListener() {
            @Override
            public void onConnect(BinaryLogClient client) {
                super.onConnect(client);
            }

            @Override
            public void onDisconnect(BinaryLogClient client) {
                super.onDisconnect(client);
            }
        });
        this.executorService.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    binaryLogClient.connect();
                } catch (Exception e) {
                    logger.warn("binaryLogClient 客户端启动失败: " + e.getMessage());
                }
            }
        });
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        if (bean.getClass().isAnnotationPresent(CdcListener.class) && bean instanceof CdcMessageListener) {
            CdcListener dbListener = bean.getClass().getAnnotation(CdcListener.class);
            this.listenerMap.put(dbListener.tableName(), beanName);
        }
        return BeanPostProcessor.super.postProcessAfterInitialization(bean, beanName);
    }

    //数据变化处理线程
    class CdcChangeThread implements Runnable {

        private String tableName;

        private String type;

        public CdcChangeThread(String tableName, String type) {
            this.tableName = tableName;
            this.type = type;
        }

        @Override
        public void run() {
            if (listenerMap.containsKey(tableName)) {
                String beanName = listenerMap.get(tableName);
                CdcMessageListener cdcMessageListener = (CdcMessageListener) applicationContext.getBean(beanName);
                cdcMessageListener.onCdcMessage(tableName, type);
            }
        }
    }
}
